package com.monitor.zk.zookeeper.handle;

import core.Request;
import com.monitor.zk.zookeeper.core.mq.Producer;
import core.annotation.HandleProcess;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.util.concurrent.LinkedBlockingQueue;

@Slf4j
/**
 * 实际发送类
 */
public class SendThreadHandleProcess extends Thread implements HandleProcess {

    public static volatile RUN_STATUS STATUS = RUN_STATUS.STOP;

    private LinkedBlockingQueue<Request> sendWork = new LinkedBlockingQueue();

    public SendThreadHandleProcess(){
        super(new HandleThreadGroup("sendThread"),"SendThreadHandleProcess");
    }

    @Override
    public void run() {
        while (true) {
            if (STATUS == RUN_STATUS.RUN) {
                try {
                    Request request = sendWork.take();
                    Producer.sendDefaultCallBack(request.getBody().toJSONString(),request.getBody().getString("type"));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (MQClientException e) {
                    e.printStackTrace();
                } catch (RemotingException e) {
                    e.printStackTrace();
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                } catch (MQBrokerException e) {
                    e.printStackTrace();
                }
            }else{
                log.warn("sendHandle未启动！");
                try {
                    Thread.sleep(10000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    @Override
    public void exec(Request request) throws InterruptedException {
        sendWork.put(request);
    }
}
